package base

import (
	"context"
	"github.com/panjf2000/ants/v2"
	"github.com/robfig/cron/v3"
	"github.com/zeromicro/go-zero/core/logx"
	"payme3000/framework/config_center"
	"payme3000/framework/errorx"
	"payme3000/framework/logxx"
	frameworkRecover "payme3000/framework/recover"
	"reflect"
	"time"
)

type (
	// ILogic 业务逻辑入口模块
	ILogic interface {
		IModule
		ISetAppConf
		ISetConfigCenter
		ISetHandleStop
		ISetAppReady
		placeholderLogic()
	}

	// Logic ILogic的基础实现
	// 注册队列消费者并维护每个消费者的生命周期
	// 注册cronjob任务并维护每个任务的生命周期
	// 注册job任务并维护每个任务的生命周期
	Logic struct {
		Module

		// Public
		StartDelay       time.Duration // 启动延迟时间
		ConsumerRegister []IConsumer   // 存放注册进来的kafka消费者
		CronJobRegister  []ICronJob    // 存放注册进来的cronjob计划任务
		JobRegister      []IJob        // 存放注册进来的job任务

		// 依赖项
		AppConf      *AppConf                    // 启动配置
		ConfigCenter *config_center.ConfigCenter // 配置中心

		// Private
		runningCronJobList  []ICronJob   // 注册成功running的cronjob
		runningJobList      []IJob       // 注册成功running的job
		runningConsumerList []IComponent // 注册成功running的kafka消费者

		cron    *cron.Cron // cronjob的驱动
		jobPool *ants.Pool // job的驱动

		handleStop func() // stop Application的handle

		isSuspend bool          // 暂停标记
		ready     chan struct{} // 就绪标记
	}
)

func (l *Logic) Start() error {
	// 默认延迟2分钟启动, k8s有30秒的就绪时间和60秒强杀等待时间
	if l.StartDelay <= 0 {
		l.StartDelay = 2 * time.Minute
	}

	// 阶段1 处理consumer消费者
	if len(l.ConsumerRegister) > 0 {
		for _, c := range l.ConsumerRegister {
			if c == nil {
				continue
			}

			// 执行consumer消费者的Init
			c.Init(l.Ctx)

			// 处理注入, 能转换成功即认为需要注入对应的实例
			// 注入 AppConf
			if _c, ok := c.(ISetAppConf); ok {
				_c.SetAppConf(l.AppConf)
			}

			// 注入 ConfigCenter
			if _c, ok := c.(ISetConfigCenter); ok {
				_c.SetConfigCenter(l.ConfigCenter)
			}

			// 注入 HandleStop
			if _c, ok := c.(ISetHandleStop); ok {
				_c.SetHandleStop(l.handleStop)
			}

			// 注入 AppReady
			if _c, ok := c.(ISetAppReady); ok {
				_c.SetAppReady(l.ready)
			}

			// 执行consumer消费者的Start, 返回error也不中断服务
			if err := c.Start(); err != nil {
				logx.Errorf("start consumer error with consumer:%s", reflect.TypeOf(c).Elem().String())
				continue
			}

			if err := l.ConfigCenter.AddConsumer(c.Topic(), c.Group(), c.Handle(), c.IsCheckDuplication(), c.IsMutex()); err != nil {
				c.Stop()
				logx.Errorf("add consumer error with err:%s, consumer:%+v", err, c)
			} else {
				l.runningConsumerList = append(l.runningConsumerList, c)
			}
		}

		// 启动消费者
		if len(l.runningConsumerList) > 0 {
			l.ConfigCenter.StartAllConsumers()
		}
	}

	// 阶段2 处理cronjob任务
	// 阶段3 处理Job任务
	go frameworkRecover.WithRecover(func() {
		select {
		case <-l.Ctx.Done(): // logic要退出
			return
		case <-time.After(l.StartDelay):
			// 阶段2 处理cronjob任务
			if len(l.CronJobRegister) > 0 {
				l.cron = cron.New() // 实例化驱动

				for _, job := range l.CronJobRegister {
					if job == nil {
						continue
					}

					_job := job
					jobType := reflect.TypeOf(_job).Elem().String()

					// init cronjob
					_job.Init(l.Ctx)

					// 处理注入, 能转换成功即认为需要注入对应的实例
					// 注入 AppConf
					if _j, ok := _job.(ISetAppConf); ok {
						_j.SetAppConf(l.AppConf)
					}

					// 注入 ConfigCenter
					if _j, ok := _job.(ISetConfigCenter); ok {
						_j.SetConfigCenter(l.ConfigCenter)
					}

					// 注入 HandleStop
					if _j, ok := _job.(ISetHandleStop); ok {
						_j.SetHandleStop(l.handleStop)
					}

					// 注入 AppReady
					if _j, ok := _job.(ISetAppReady); ok {
						_j.SetAppReady(l.ready)
					}

					// start cronjob, 返回error也不中断服务
					if err := _job.Start(); err != nil {
						logx.Errorf("start cronjob error with err:%s, cronjob:%s, key:%s", err, jobType, _job.getKey())
						continue
					}

					logx.Infof("start cronjob done with jobType:%s", jobType)

					// 注册cronjob任务
					_job.setCron(l.cron)
					if entryId, err := l.cron.AddFunc(
						// 周期频率
						_job.Spec(),
						// 时间到要执行的回调
						func() {
							select {
							case <-l.Ctx.Done(): // logic要退出
							default:
								<-l.ready // 等待app完全就绪

								// 执行调度
								_job.schedule(logxx.NewLogContext(), func(ctx context.Context) {
									_job.Run(ctx) // 自定义的任务逻辑
								})
							}
						},
					); err != nil {
						logx.Errorf("register cronjob error with err:%s", err)
					} else {
						_job.setEntryId(entryId)

						l.runningCronJobList = append(l.runningCronJobList, _job)

						logx.Infof("register cronjob done with jobType:%s, jobKey:%s, entryId:%v", jobType, _job.getKey(), entryId)
					}
				}

				// 启动cronjob驱动, 开始调度
				l.cron.Start()
			}

			// 阶段3 处理Job任务
			if len(l.JobRegister) > 0 {
				jobPool, err := ants.NewPool(len(l.JobRegister), ants.WithNonblocking(true))
				if err != nil {
					errorx.Catch(context.Background(), err)
					return
				} else {
					l.jobPool = jobPool
				}

				for _, job := range l.JobRegister {
					if job == nil {
						continue
					}

					_job := job
					jobType := reflect.TypeOf(_job).Elem().String()

					// init cronjob
					_job.Init(l.Ctx)

					// start job, 返回error也不中断服务
					if err := _job.Start(); err != nil {
						logx.Errorf("start job error with err:%s, jobType:%s, jobKey:%s", err, jobType, _job.getKey())
						continue
					}

					// 提交到协程池
					if err := jobPool.Submit(func() {
						select {
						case <-l.Ctx.Done(): // logic要退出
						default:
							<-l.ready // 等待app完全就绪

							// 执行调度
							_job.schedule(logxx.NewLogContext(), func(ctx context.Context) {
								_job.Run(ctx) // 自定义的任务逻辑
							})
						}
					}); err != nil {
						errorx.Catch(context.Background(), err)
					} else {
						l.runningJobList = append(l.runningJobList, _job)

						logx.Infof("register job done with jobType:%s, jobKey:%s", jobType, _job.getKey())
					}
				}

				// 设置了当所有job调度完成就退出程序
				if l.AppConf.IsStopAfterJobDone {
					go frameworkRecover.WithRecover(func() {
						for _, job := range l.runningJobList {
							job.wait()
						}

						// 待全部Job都调度后, stop Application
						l.handleStop()

						logx.Info("handle stop done")
					})
				}
			}
		}
	})

	return nil
}

func (l *Logic) Stop() {
	// 先停止cronjob和consumer的驱动
	// 停止调度cronjob
	if l.cron != nil {
		// 最多等待10秒
		select {
		case <-l.cron.Stop().Done():
		case <-time.After(10 * time.Second):
		}
	}

	// 停止kafka消费
	if len(l.runningConsumerList) > 0 {
		l.ConfigCenter.StopAllConsumers()
	}

	// stop all cronjob
	for _, j := range l.runningCronJobList {
		j.Stop()
	}

	// stop all job
	for _, j := range l.runningJobList {
		j.Stop()
	}

	// stop all consumer
	for _, c := range l.runningConsumerList {
		c.Stop()
	}

	// 释放job协程池
	if l.jobPool != nil {
		l.jobPool.Release()
	}

	// 覆盖了默认的Stop, 手动执行默认Stop
	l.Module.Stop()
}

func (l *Logic) Suspend() {
	// cronjob的驱动不需要暂停, cronjob的默认Suspend已经实现Suspend功能, 同理Resume
	// kafka需要把驱动暂停了, consumer实现的Suspend并不控制kafka的驱动暂停消费, 同理Resume
	if len(l.runningConsumerList) > 0 {
		l.ConfigCenter.StopAllConsumers()
	}

	for _, c := range l.runningCronJobList {
		c.Suspend()
	}

	for _, c := range l.runningConsumerList {
		c.Suspend()
	}
}

func (l *Logic) Resume() {
	for _, c := range l.runningCronJobList {
		c.Resume()
	}

	for _, c := range l.runningConsumerList {
		c.Resume()
	}

	// 启动kafka消费
	if len(l.runningConsumerList) > 0 {
		l.ConfigCenter.StartAllConsumers()
	}
}

func (l *Logic) SetAppConf(appConf *AppConf) {
	l.AppConf = appConf
}

func (l *Logic) SetConfigCenter(configCenter *config_center.ConfigCenter) {
	l.ConfigCenter = configCenter
}

func (l *Logic) SetHandleStop(handle func()) {
	l.handleStop = handle
}

func (l *Logic) SetAppReady(ready chan struct{}) {
	l.ready = ready
}

func (l *Logic) placeholderLogic() {}
